package coutom.sink;

import coutom.sink.bean.Person;
import coutom.sink.config.SinkConfig;
import coutom.sink.utils.excel.ExcelUtils;
import jxl.Workbook;
import jxl.write.WritableSheet;
import jxl.write.WritableWorkbook;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;

/**
 * @author : ZhaoY
 * @description : 将消息队列的数据写入到file中
 * @date : 2021-02-25 15:50
 **/
@Slf4j
public class FileSink implements Sink<byte[]> {
    private BufferedWriter bw = null;
    private FileWriter fw = null;
    private SinkConfig sinkConfig = null;
    private File file = null;

    @Override
    public void open(Map<String, Object> map, SinkContext sinkContext) throws Exception {
//        对yaml配置文件进行装载
        sinkConfig = SinkConfig.load(map);
//        创建Excel文件
        file = new File(sinkConfig.getOutDirectory() + File.separator + sinkConfig.getFileName());
        WritableWorkbook workbook = null;
        workbook = Workbook.createWorkbook(file);
        WritableSheet ws = workbook.createSheet("mysheet", 0);
        //设置各列的长度
        ws.setColumnView(0, 30);
        ws.setColumnView(1, 30);
        ws.setColumnView(2, 30);
        //写入到文件
        workbook.write();
        //释放资源
        workbook.close();
    }

    @Override
    public void write(Record<byte[]> record) throws Exception {
        String result = new String(record.getValue());
        //消息格式：75,Zhaoy,124132017075
        result = result.replaceAll("\r|\n", "");
        String[] pArray = result.split(",");
        Person person = new Person(Integer.valueOf(pArray[0]), pArray[1], pArray[2]);
        Boolean flag = ExcelUtils.updateOneExcel(file, person);
    }

    @Override
    public void close() {
        try {
            if (bw != null)
                bw.close();
            if (fw != null)
                fw.close();
        } catch (IOException ex) {
            log.info("Excel资源流关闭异常");
        }
    }

}
